package org.raftkv.raft.core.log;

import org.raftkv.commons.Tuple;
import org.raftkv.raft.exceptions.RaftException;
import org.raftkv.raft.pb.Entry;
import org.raftkv.raft.pb.Snapshot;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * unstable数据结构用于还没有被用户层持久化的数据，而其中又包括两部分：快照数据和日志条目数组
 * 这两个部分，并不同时存在，同一时间只有一个部分存在。
 * 快照数据仅当当前节点在接收从leader发送过来的快照数据时存在，在接收快照数据的时候，
 * entries数组中是没有数据的；除了这种情况之外，就只会存在entries数组的数据了。
 */
public class Unstable {

    private static final Logger LOGGER = LoggerFactory.getLogger(Unstable.class);

    /**
     * 保存还没有持久化的快照数据
     * 在启动之后同步数据的过程中，如果需要同步快照数据时才会去进行赋值修改的数据
     */
    private Snapshot snapshot;

    /**
     * 保存的是entries数组中的第一条数据在raft日志中的索引，
     * 即第i条entries数据在raft日志中的索引为i + offset。
     */
    private long offset;

    /**
     * 还未持久化的数据
     */
    private List<Entry> entries;

    public Unstable() {
        this.entries = new ArrayList<>();
    }

    public Unstable(Snapshot snapshot, List<Entry> entries, long offset) {
        this.snapshot = snapshot;
        this.entries = new ArrayList<>(entries);
        this.offset = offset;
    }

    public Unstable(long offset) {
        this.offset = offset;
    }

    /**
     * 只有在快照存在的情况下，返回快照中的meta数据
     * 注意到与取lastIndex值不同的是，firstindex只会从快照中尝试。
     * 这是因为快照中的数据用于刚启动时，此时可能是最早的index，
     * 而entry数组存在数据时，持久化中已经有数据了，此时不可能从entry数组中取firstindex
     */
    public Tuple<Long, Boolean> maybeFirstIndex() {
        if (snapshot != null) {
            return new Tuple<>(snapshot.getMetadata().getIndex() + 1, true);
        }
        return new Tuple<>(0L, false);
    }

    /**
     * 如果entries存在，返回最后一个entry的索引
     * 否则如果快照存在，返回快照的meta数据中的索引
     * 以上都不成立，则返回0
     */
    public Tuple<Long, Boolean> maybeLastIndex() {
        int l = entries == null ? 0 : entries.size();
        if (l != 0) {
            return new Tuple<>(offset + l - 1, true);
        }
        if (snapshot != null) {
            return new Tuple<>(snapshot.getMetadata().getIndex(), true);
        }
        return new Tuple<>(0L, false);
    }

    /**
     * 返回索引i的term，如果不存在则返回0
     *
     * @param i 索引
     * @return 索引对应的日志的term
     */
    public Tuple<Long, Boolean> maybeTerm(long i) {
        if (i < offset) {
            if (snapshot != null && snapshot.getMetadata().getIndex() == i) {
                return new Tuple<>(snapshot.getMetadata().getTerm(), true);
            }
            return new Tuple<>(0L, false);
        }
        Tuple<Long, Boolean> t = maybeLastIndex();
        Long last = t.getFirst();
        Boolean ok = t.getSecond();
        if (!ok) {
            return new Tuple<>(0L, false);
        }
        //如果比lastIndex还大，那也查不到
        if (i > last) {
            return new Tuple<>(0L, false);
        }
        return new Tuple<>(entries.get((int) (i - offset)).getTerm(), true);
    }

    /**
     * 传入索引i和term，表示目前这块数据已经持久化了
     *
     * @param i 索引
     * @param t 任期
     */
    public void stableTo(long i, long t) {
        Tuple<Long, Boolean> temp = maybeTerm(i);
        Long gt = temp.getFirst();
        Boolean ok = temp.getSecond();

        if (!ok) {
            return;
        }
        //只有在term相同，同时索引大于等于当前offset的情况下
        if (gt == t && i >= offset) {
            // 因为前面的数据被持久化了，所以将entries缩容，从i开始
            entries = new ArrayList<>(entries.subList((int) (i + 1 - offset), entries.size()));
            offset = i + 1;
        }
    }

    public void stableSnapTo(long i) {
        if (snapshot != null && snapshot.getMetadata().getIndex() == i) {
            snapshot = null;
        }
    }

    public void restore(Snapshot s) {
        offset = s.getMetadata().getIndex() + 1;
        entries = null;
        snapshot = s;
    }

    /**
     * 传入entries，可能会导致原先数据的截断或者添加操作
     * 这就是最开始注释中说明的offset可能比持久化索引小的情况，需要做截断
     *
     * @param ents 需要添加的entries
     */
    public void truncateAndAppend(List<Entry> ents) {
        long after = ents.get(0).getIndex();
        if (after == offset + (this.entries == null ? 0 : this.entries.size())) {
            // 如果正好是紧接着当前数据的，就直接append
            if (this.entries == null) {
                this.entries = new ArrayList<>(ents);
            } else {
                this.entries.addAll(ents);
            }
        } else if (after <= offset) {
            LOGGER.info("replace the unstable entries from index {}", after);
            // 如果比当前偏移量小，那用新的数据替换当前数据，需要同时更改offset和entries
            offset = after;
            this.entries = ents;
        } else {
            // 到了这里，说明 u.offset < after < u.offset+uint64(len(u.entries))
            // 那么新的entries需要拼接而成
            LOGGER.info("truncate the unstable entries before index {}", after);
            this.entries = new ArrayList<>(slice(offset, after));
            this.entries.addAll(ents);
        }
    }

    /**
     * @param lo 左区间，包括
     * @param hi 右区间，不包括
     * @return 返回索引范围在[lo-u.offset : hi-u.offset)之间的数据
     */
    public List<Entry> slice(long lo, long hi) {
        checkOutOfBounds(lo, hi);
        return new ArrayList<>(entries.subList((int) (lo - offset), (int) (hi - offset)));
    }

    private void checkOutOfBounds(long lo, long hi) {
        if (lo > hi) {
            String errMsg = String.format("invalid unstable.slice %d > %d", lo, hi);
            LOGGER.error(errMsg);
            throw new RaftException(errMsg);
        }

        long upper = offset + entries.size();
        if (lo < offset || hi > upper) {
            String errMsg = String.format("unstable.slice[%d,%d) out of bound [%d,%d]", lo, hi, offset, upper);
            LOGGER.error(errMsg);
            throw new RaftException(errMsg);
        }
    }

    public Snapshot getSnapshot() {
        return snapshot;
    }

    public void setSnapshot(Snapshot snapshot) {
        this.snapshot = snapshot;
    }

    public List<Entry> getEntries() {
        return entries;
    }

    public void setEntries(List<Entry> entries) {
        this.entries = entries;
    }

    public long getOffset() {
        return offset;
    }

    public void setOffset(long offset) {
        this.offset = offset;
    }

}
